package com.spark;

import com.trs.hybase.client.TRSConnection;
import com.trs.hybase.client.TRSDatabase;
import com.trs.hybase.client.TRSException;
import com.trs.hybase.client.TRSUtils;
import com.trs.hybase.client.params.ConnectParams;


import java.io.IOException;
import java.util.HashMap;

/**
 * Created by Administrator on 2017/7/7.
 */
public class HybaseTest {
    public static void main(String[] args) throws TRSException, IOException {


        TRSConnection conn = new TRSConnection("http://hybase.zhaochao.net", "admin", "Trsadmin_2015_!@#", new ConnectParams());


        String query="(DOCPUBTIME:[\"2017.07.10 00:00:00\" TO \"2017.07.10 15:19:02\"] AND ZB_FINAL_AREA:001020 AND TXS:[400 TO * ] AND IR_MIMETYPE#LIST:application\\/xml,application\\/json,application\\/javascript,text\\/html AND ZB_TYPE:wz) OR (DOCPUBTIME:[\"2017.07.10 00:00:00\" TO \"2017.07.10 15:19:02\"] AND ZB_FINAL_AREA:001020 AND TXS:[400 TO * ] AND  ZB_TYPE:jtcpg AND MC#LIST:浙江在线网站,浙江新闻APP,浙江日报,钱江晚报,钱江晚报公众号) OR (DOCPUBTIME:[\"2017.07.10 00:00:00\" TO \"2017.07.10 15:19:02\"] AND ZB_FINAL_AREA:001020 AND TXS:[400 TO * ] NOT ZB_TYPE#LIST:jtcpg,wz ) NOT DOCTITLE:\"无标题\"\n";
        String mainDb="website_formal_20160321;app_formal_20160321;weixin_formal_20160321;mlf_product_formal_20160511";


        HashMap e = new HashMap();
        e.put("list.db.splitted", "true");
        TRSDatabase[] databases = conn.getDatabases(mainDb, e);
        e = new HashMap();
        e.put("dbname", mainDb);
        e.put("query", query);
        String[] arrSubInfos = conn.executeCommand("hadoop_getsplits", e).split("\r\n");
        String[] arr$ = arrSubInfos;
        int len$ = arrSubInfos.length;

        for(int i$ = 0; i$ < len$; ++i$) {
            String subInfo = arr$[i$];
            if(subInfo != null && subInfo.length() != 0) {
                HashMap opts = TRSUtils.parseProperties(subInfo);
                String host = (String)opts.get("host");
                String port = (String)opts.get("port");
                String dbname = (String)opts.get("dbname");
                String subdb = (String)opts.get("subdb");
                long recTotal = Long.parseLong((String)opts.get("recnum"));
                long batchNum = (long)Math.ceil((double)(0.01F + 1.0F * (float)recTotal / 1.0E7F));
                batchNum = Math.min(batchNum, 5L);
                long batchRecords = 1L + recTotal / batchNum;
                long batchNo = 0L;

                for(long readPos = 0L; batchNo < batchNum && readPos < recTotal; readPos += batchRecords) {
                    long readNum = readPos + batchRecords < recTotal?batchRecords:recTotal - readPos;

                    System.out.println(host+ port+ dbname+ subdb+ query+ "\n"+ ""+ readPos+ readNum+ recTotal+ databases[0]);

                    ++batchNo;
                }
            }
        }

    }
}
